package com.example.hadoop.mapreduce.logenhance;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Auther: Brian
 * @Date: 2020/04/28/22:59
 * @Description: 自定义OutputFormat
 * map task 或 reduce task在最终输出是，会先调用OutputFormat的getRecodeWriter(）方法拿到一个RecordWriter
 * 再调用RecordWriter的writer(k,v) 方法将数据写出
 */
//Key-Value 就是map task 或reduce task给到的Key-Value
public class LogEnhanceOutputFormat extends FileOutputFormat<Text, NullWritable> {

    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        FileSystem fs = FileSystem.get(context.getConfiguration());
        FSDataOutputStream enhanceOS = fs.create(new Path("/logenhance/enhancelog/enhancelog.data"));
        FSDataOutputStream toCrawlOS = fs.create(new Path("/logenhance/tocrawllog/tocrawl.data"));
        return new LogEnhanceRecordWriter(enhanceOS, toCrawlOS);
    }

   static class LogEnhanceRecordWriter extends RecordWriter<Text, NullWritable> {
        FSDataOutputStream enhanceOS = null;
        FSDataOutputStream toCrawlOS = null;

        public LogEnhanceRecordWriter(FSDataOutputStream enhanceOS, FSDataOutputStream toCrawlOS) {
            super();
            this.enhanceOS = enhanceOS;
            this.toCrawlOS = toCrawlOS;
        }

        /**
         * 自定义Record Writer， 数据将调用write方法进行输出数据
         * 这里需要输出不同的路径下，所有需要传多个OutputFormat进来
         *
         * @param key
         * @param value
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void write(Text key, NullWritable value) throws IOException, InterruptedException {
            String line = key.toString();
            if (line.contains("tocrawl")) {
                //待爬的url
                toCrawlOS.write(line.getBytes());
            } else {
                //输出到增强日志目录
                enhanceOS.write(line.getBytes());
            }
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            if (enhanceOS != null) {
                enhanceOS.close();
            }
            if (toCrawlOS != null) {
                toCrawlOS.close();
            }
        }
    }
}
